package xuecheng.tool.mq.alibaba.rocket;

import java.util.Map;

import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import xuecheng.tool.json.JsonUtils;
import xuecheng.tool.json.JsonUtilsFactory;

/**
 * 阿里云 RocketMq 工具类
 *
 * @author PengXueCheng
 * @version 1.0.0
 */
@Slf4j
public class RocketMqUtils {

    private final JsonUtils jsonUtils = JsonUtilsFactory.obtain();
    /**
     * 设置 TCP 接入域名，进入控制台的实例管理页面，在页面上方选择实例后，在实例信息中的“获取接入点信息”区域查看 默认
     */
    private final String nameServerAddress;
    /**
     * 生产者集合
     */
    private final Map<String, Producer> producers;

    public RocketMqUtils(Map<String, Producer> producers, @NonNull String nameServer) {
        this.producers = producers;
        this.nameServerAddress = nameServer;
    }

    /**
     * 发送消息
     *
     * @param nameServerAddress
     *            设置 TCP 接入域名，进入控制台的实例管理页面，在页面上方选择实例后，在实例信息中的“获取接入点信息”区域查看
     * @param gid
     *            您在控制台创建的 Group ID
     * @param topic
     *            消息一级分类
     * @param tag
     *            消息二级分类(可选, 默认为 *)
     * @param key
     *            设置代表消息的业务关键属性，请尽可能全局唯一，以方便您在无法正常收到消息情况下，可通过控制台查询消息并补发. 注意：不设置也不会影响消息正常收发
     * @param message
     *            消息
     * @param <T>
     *            消息实体类型
     */
    public <T> void send(@NonNull String nameServerAddress, @NonNull String gid, @NonNull String topic,
        @Nullable String tag, @Nullable String key, @NonNull T message) {
        if (StringUtils.isEmpty(tag)) {
            tag = "*";
        }

        String producerKey = String.format("%s-%s", nameServerAddress, gid);
        Producer producer = producers.get(producerKey);
        assert producer != null;

        String messageStr = jsonUtils.toJson(message);

        Message msg = new Message(
            // 在控制台创建的 Topic，即该消息所属的 Topic 名称
            topic,
            // Message Tag,
            // 可理解为 Gmail 中的标签，对消息进行再归类，方便 Consumer 指定过滤条件在消息队列 RocketMQ 服务器过滤
            tag,
            // Message Body
            // 任何二进制形式的数据， 消息队列 RocketMQ 不做任何干预，
            // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
            messageStr.getBytes());

        // 设置代表消息的业务关键属性，请尽可能全局唯一，以方便您在无法正常收到消息情况下，可通过控制台查询消息并补发
        // 注意：不设置也不会影响消息正常收发
        msg.setKey(key);
        // 发送消息，只要不抛异常就是成功
        // 打印 Message ID，以便用于消息发送状态查询
        SendResult sendResult;
        try {
            sendResult = producer.send(msg);
        } catch (ONSClientException e) {
            log.error(producer.toString() + "_" + msg + "_" + e.getMessage(), e);
            return;
        }
        log.info("Send Message success. topic:{}, MessageID:{}, producerGroup:{}", String.format("%s:%s", topic, tag),
            sendResult.getMessageId(), gid);
    }

    /**
     * 发送消息
     *
     * @param nameServerAddress
     * @param gid
     * @param topic
     * @param message
     * @param <T>
     */
    public <T> void send(@NonNull String nameServerAddress, @NonNull String gid, @NonNull String topic,
        @NonNull T message) {
        send(nameServerAddress, gid, topic, null, null, message);
    }

    /**
     * 发送消息
     *
     * @param gid
     * @param topic
     * @param message
     * @param <T>
     */
    public <T> void send(@NonNull String gid, @NonNull String topic, @NonNull T message) {
        send(nameServerAddress, gid, topic, null, null, message);
    }

}